package com.atguigu.flink.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.edu.common.GmallConfig;
import com.atguigu.flink.edu.utils.DataFormatUtil;
import com.atguigu.flink.edu.utils.MyKafkaUtil;
import com.atguigu.flink.edu.utils.cartAddUuBean;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;

public class DwsTradeCartAddUuWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1:基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        ////TODO 2:开启检查点
        //env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        //env.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.days(30), org.apache.flink.api.common.time.Time.seconds(3)));
        ////设置检查点超时时间
        //env.getCheckpointConfig().setCheckpointTimeout(6000L);
        ////设置检查点最小间隔时间
        //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        ////设置取消job是否保存检查点
        //env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        ////设置状态后端
        //env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop001:8020/edu/ck");
        ////设置hadoop操作用户
        //System.setProperty("HADOOP_USER_NAME","atguigu");
        //TODO 3:从kafka中读取加购数据
        //3.1 创建表对象
        //String topic = "dwd_trade_cart_add";
        //String groupId = "dws_trade_cart_add_uu";
        //KafkaSource<String> kafkaStr = MyKafkaUtil.getKafkaSource(topic, groupId);
        ////3.2将数据转换为流
        //DataStreamSource<String> sourceStr
        //        = env.fromSource(kafkaStr, WatermarkStrategy.noWatermarks(), "kafka_source");
        ////3.3对数据进行类型转换
        //SingleOutputStreamOperator<JSONObject> jsonObj = sourceStr.map(JSON::parseObject);
        String topic = "dwd_cart_info";
        String groupId = "dws_trade_cart_add_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
//3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaStrDS.map(JSON::parseObject);
        //{"sku_num":"1","user_id":"105","sku_id":"13","source_type":"2401","id":"3408","source_type_name":"用户查询","ts":"1654046562"}
        //jsonObjDS.print(">>>>>");
        //jsonObj.print();
        //TODO 4:插入水位线,并且指定事件时间字段
        //jsonObj.assignTimestampsAndWatermarks(
        //        WatermarkStrategy
        //                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
        //                .withTimestampAssigner(
        //                        new SerializableTimestampAssigner<JSONObject>() {
        //                            @Override
        //                            public long extractTimestamp(JSONObject jsonObject, long l) {
        //                                return jsonObject.getLong("ts") * 1000L;
        //                            }
        //                        }
        //                )
        //);
        SingleOutputStreamOperator<JSONObject> watermarksDs = jsonObj.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );
        //TODO 5:对userID进行分组
        KeyedStream<JSONObject, String> keyedDs
                = watermarksDs.keyBy(jsonObj1 -> jsonObj1.getString("user_id"));
        //KeyedStream<JSONObject, String> keyedDs = jsonObj.keyBy(r -> r.getString("user_id"));
        //TODO 6:使用状态编程,过滤出独立用户
        //SingleOutputStreamOperator<JSONObject> processDs
        //        = keyedDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
        //    private ValueState<String> valueState;
//
        //    @Override
        //    public void open(Configuration parameters) throws Exception {
        //        ValueStateDescriptor<String> value_state = new ValueStateDescriptor<>("value_state", String.class);
        //        this.valueState = getRuntimeContext().getState(value_state);
        //        value_state.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build());
//
        //    }
//
        //    @Override
        //    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
        //        String value = valueState.value();
        //        String currData = DataFormatUtil.toDate(jsonObject.getLong("ts") * 1000);
        //        if (StringUtils.isEmpty(value) || !value.equals(currData)) {
        //            collector.collect(jsonObject);
        //            valueState.update(currData);
        //        }
        //    }
        //});
        SingleOutputStreamOperator<JSONObject> processDs = keyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> lastCartAddDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastCartAddDateState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build());
                        this.lastCartAddDateState
                                = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        String lastCartAddDate = lastCartAddDateState.value();

                        Long ts = jsonObj.getLong("ts") * 1000;
                        String curCartAddDate = DataFormatUtil.toDate(ts);

                        if (StringUtils.isEmpty(lastCartAddDate) || !lastCartAddDate.equals(curCartAddDate)) {
                            out.collect(jsonObj);
                            lastCartAddDateState.update(curCartAddDate);
                        }
                    }
                }
        );
        //TODO 7:开窗
        //processDs.print("==>");
        AllWindowedStream<JSONObject, TimeWindow> windowDs = processDs.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //聚合
        //SingleOutputStreamOperator<cartAddUuBean> aggDs = windowDs.aggregate(
        //        new AggregateFunction<JSONObject, Long, Long>() {
        //            @Override
        //            public Long createAccumulator() {
        //                return 0L;
        //            }
//
        //            @Override
        //            public Long add(JSONObject jsonObject, Long acc) {
        //                return ++acc;
        //            }
//
        //            @Override
        //            public Long getResult(Long acc) {
        //                return acc;
        //            }
//
        //            @Override
        //            public Long merge(Long aLong, Long acc1) {
        //                return null;
        //            }
        //        }
        //        ,
        //        new AllWindowFunction<Long, cartAddUuBean, TimeWindow>() {
        //            @Override
        //            public void apply(TimeWindow timeWindow, Iterable<Long> iterable, Collector<cartAddUuBean> collector) throws Exception {
        //                for (Long value : iterable) {
        //                    collector.collect(
        //                            new cartAddUuBean(
        //                                    DataFormatUtil.toDate(timeWindow.getStart()),
        //                                    DataFormatUtil.toDate(timeWindow.getEnd()),
        //                                    value,
        //                                    System.currentTimeMillis()
        //                            )
        //                    );
        //                }
        //            }
        //        }
        //);
        //TODO 9.聚合计算
        SingleOutputStreamOperator<cartAddUuBean> aggDS = windowDs.aggregate(
                new AggregateFunction<JSONObject, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(JSONObject value, Long accumulator) {
                        return ++accumulator;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                },
                new AllWindowFunction<Long, cartAddUuBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<Long> values, Collector<cartAddUuBean> out) throws Exception {
                        for (Long value : values) {
                            out.collect(new cartAddUuBean(
                                    DataFormatUtil.toYmdHms(window.getStart()),
                                    DataFormatUtil.toYmdHms(window.getEnd()),
                                    value,
                                    System.currentTimeMillis()
                            ));
                        }
                    }
                }
        );
        aggDS.print();
        //将数据写入CK
        aggDS.addSink(JdbcSink.sink("insert into dws_trade_cart_add_uu_window values(?,?,?,?)",
                new JdbcStatementBuilder<cartAddUuBean>(){
                    @Override
                    public void accept(PreparedStatement ps, cartAddUuBean cartAddUuBean) throws SQLException {
                        ps.setObject(1,cartAddUuBean.getStt());
                        ps.setObject(2,cartAddUuBean.getEdt());
                        ps.setObject(3,cartAddUuBean.getCartAddUuCt());
                        ps.setObject(4,cartAddUuBean.getTs());
                    }
                },
                new JdbcExecutionOptions
                        .Builder()
                        .withBatchIntervalMs(3000)
                        .withBatchSize(1)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        ));
        env.execute();
    }
}